package com.flink;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class TimeWindow {

    /**
     * 时间语义：
     * eventTime需要设置：
     * 1. 环境时间语义；
     * 2. 指定时间字段
     *
     * @param environment
     */
    protected static void singlePreKeyByEventTimeTumblingWindow(StreamExecutionEnvironment environment) {
        // 指定时间语义
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> socketTextStream = environment.socketTextStream("localhost", 1234);
        // 指定Source指定时间字段
        SingleOutputStreamOperator<String> dataWithOriginWater = socketTextStream.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner((SerializableTimestampAssigner<String>) (originalData, l) -> {
            String[] split = originalData.split(",");
            long time = l;
            try {
                time = Long.parseLong(split[0]);
            } catch (Exception e) {
            }
            return time;
        }));

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> mapOperator = dataWithOriginWater.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String originalData) throws Exception {
                String[] split = originalData.split(",");
                return Tuple3.of(split[0], split[1], 1);
            }
        });

        KeyedStream<Tuple3<String, String, Integer>, String> keyedOperator = mapOperator.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> stringStringTuple2) throws Exception {
                return stringStringTuple2.f1;
            }
        });

        WindowedStream<Tuple3<String, String, Integer>, String, org.apache.flink.streaming.api.windowing.windows.TimeWindow> window = keyedOperator.window(TumblingEventTimeWindows.of(Time.seconds(5), Time.seconds(0)));

        window.sum(2).print();
    }


}
